#include <sys/socket.h>
#include <sys/epoll.h>
#include <errno.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdio.h>

#include "conn.h"
#include "interface.h"
#include "debug.h"
#include "help.h"

extern ctx_t g_ctx;


extern int sock_general_init (int sockfd);


void add_event(conn_t* conn){
	struct epoll_event ev;
	
	ev.data.ptr = (void*)conn;
	ev.events = EPOLLIN | EPOLLOUT| EPOLLET;
	epoll_ctl(g_ctx.epfd, EPOLL_CTL_ADD, conn->sockfd, &ev);
}

void del_event(conn_t* conn){
	struct epoll_event ev;
	
	ev.data.ptr = (void*)conn;
	ev.events = EPOLLIN | EPOLLOUT| EPOLLET;
	epoll_ctl(g_ctx.epfd, EPOLL_CTL_DEL, conn->sockfd, &ev);
}

static rval_t handle_accept(){
		conn_t* conn ;
		struct sockaddr_in cli_addr;
		int len;
		int connfd;
		rval_t rval;
		
		int node_idx = -1;

		rval.v = 0;
		
		connfd = accept(g_ctx.listenfd, (struct sockaddr *) &cli_addr, &len);
		if (connfd  == -1){
				debug_err("accept failed! errno:%d\n", errno);
				assert(connfd >0);
				rval.v= -1;
				rval.err = errno;
				return  rval;
		}

		setblock(connfd,0);
		sock_general_init(connfd);

		debug_info("accept new conn on  fd[%d], addr:%s\n", connfd, dump_addr(cli_addr));

		
        node_idx = find_node_from_addr(cli_addr);
		assert(node_idx!=-1);
		conn =	conn_tab_find_from_node(&g_ctx.conf.node[node_idx]);
		if(conn){
			conn->sockfd = connfd; 
			conn->addr = cli_addr;
			//assert(conn->conn_type == ACCEPT);
			assert(conn->conn_state == CONNECTING);
		}else{
			conn = create_conn(node_idx, connfd, cli_addr);
			conn->conn_type =ACCEPT;
			conn_tab_add(conn);
		}
		
		set_conn_state(conn, CONNECTED);
		
		add_event(conn);
	
		return rval;
}


extern void* send_fun(void* arg);

void handle_out(struct epoll_event*  ev){
		conn_t*  conn;
		task_t* task;

		conn=  (conn_t*)ev->data.ptr;
		assert(conn!=NULL);

		debug_info("epoll out event on conn[%d], fd[%d], addr:%s\n ", 
			            conn->id, conn->sockfd, dump_addr(conn->addr));

		/*
		pthread_mutex_lock(&conn->lock);
			if(conn->snd_state == IDEL){ //not in  thread task queue
				task = new_task(send_fun, (void*)conn);
				assert(task!=NULL);

	            debug_info("new send task:%llu, conn[%d], conn sndstate[%d]\n",
						 task->id, conn->id, conn->snd_state);

				schedule(&g_ctx.tp, task);
				conn->snd_state = SND_READY;
			}else{
				debug_info("conn is not in idel, not create new task!"
						   " conn[%d], conn snd_state[%d]\n", conn->id, conn->snd_state);
			}
		pthread_mutex_unlock(&conn->lock);
        */


		task = new_task(send_fun, (void*)conn);
		 assert(task!=NULL);

	          debug_info("new send task:%llu, conn[%d], conn sndstate[%d]\n",
						 task->id, conn->id, conn->snd_state);

		 schedule(&g_ctx.tp, task);
 		
		//ev.data.fd = connfd;
		//ev.events = EPOLLIN | EPOLLET;
		//epoll_ctl(g_ctx.epfd, EPOLL_CTL_MOD, connfd, &ev);

		
}

extern void* recv_fun(void* arg);

void handle_in(struct epoll_event*  ev){
	conn_t*  conn;
	task_t* task;

	
	conn=  (conn_t*)ev->data.ptr;
			assert(conn!=NULL);
	
	debug_info("epoll in event on conn[%d], fd[%d], addr:%s\n ", 
				conn->id, conn->sockfd, dump_addr(conn->addr));
	
/*
	pthread_mutex_lock(&conn->lock);
			if(conn->rcv_state == IDEL){
				task =  new_task(recv_fun, (void*)conn);
				debug_info("new recv task:%llu\n", task->id);
				schedule(&g_ctx.tp, task);
				conn->rcv_state = RCV_READY;
			}
	pthread_mutex_unlock(&conn->lock);*/


	task =  new_task(recv_fun, (void*)conn);
	debug_info("new recv task:%llu\n", task->id);
	schedule(&g_ctx.tp, task);
	
	
}
void*  epoll_fun(void* arg){
		int nfds;
		int i;
		int timeout = -1;   //never timeout
		struct epoll_event  events[MAX_EVENT];

		while(1){
			nfds = epoll_wait(g_ctx.epfd, events, MAX_EVENT, timeout);
			if(nfds == -1){
				debug_info("epoll wait failed! errno:%d\n", errno);
				break;
			}	
			for (i = 0; i < nfds;  i++) {
					if (events[i].data.fd == g_ctx.listenfd){
							handle_accept();

					}else if (events[i].events & EPOLLIN){ //receive
							handle_in(&events[i]);

					}else if (events[i].events & EPOLLOUT){ //send
							handle_out(&events[i]);
					}else if (events[i].events & EPOLLERR){
						conn_t* conn = (conn_t*)events[i].data.ptr;
						debug_warn("epoll epoller event!\n");
						add_reconn(conn);
					}else if(events[i].events & EPOLLHUP){
						conn_t* conn = (conn_t*)events[i].data.ptr;		
						debug_warn("epoll epollhup event!\n");
						add_reconn(conn);
					}
			}
		}
}


void start_epoll_thread(){
		int re;
		re = pthread_create(&g_ctx.listen_pid, NULL, epoll_fun, NULL);
		assert(re==0);
		debug_info("start epoll thread success!\n");
}

void stop_epoll_thread(){
	pthread_join(g_ctx.listen_pid, NULL);
}


